package com.telra.app.server2.service;

import com.telra.app.server2.config.CassandraConfig;
import com.telra.app.server2.config.MQConfig;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;

/**
 * Connects to Zookeeper to load configuration information for all the services
 */
public class ZookeeperService implements Watcher {
    
    private static final Log4JLogger log = (Log4JLogger) LogFactory.getFactory().getInstance(ZookeeperService.class);

    private ZooKeeper zk;

    private String hostname;
    private String port;
    private int timeout;

    String cassandraHostPath;   //path to cassandra host name in ZK
    String cassandraPortPath;   //path to cassandra port in ZK
    String cassandraClusterPath;   //path to cassandra cluster in ZK
    String cassandraKeyspacePath;  //path to cassandra keyspace in ZK
    String mqUrlPath;    //path to MQ URL path in ZK
    String mqTopicPath;  //path to MQ topic in ZK

    //loaded from ZK
    private CassandraConfig cassandraConfig;

    private MQConfig mqConfig;

    public ZookeeperService() {
        //empty
    }

    public ZooKeeper getZooKeeper() {
        return zk;
    }

    public void init() {
        try {
            log.info("Initializing ZK");
            zk = new ZooKeeper(hostname+":" + port, timeout, this);

            cassandraConfig = new CassandraConfig();
            cassandraConfig.setCassandraHost(setConfigEntries(cassandraHostPath));
            cassandraConfig.setCassandraPort(setConfigEntries(cassandraPortPath));
            cassandraConfig.setCassandraCluster(setConfigEntries(cassandraClusterPath));
            cassandraConfig.setCassandraKeyspace(setConfigEntries(cassandraKeyspacePath));

            mqConfig = new MQConfig();
            mqConfig.setMqUrl(setConfigEntries(mqUrlPath));
            mqConfig.setMqTopic(setConfigEntries(mqTopicPath));
        }
        catch (IOException e1) {
            log.error(e1);
            e1.printStackTrace();
        }
    }

    private String setConfigEntries(String zkPath) {
        String zkValue = null;

        try {
            Stat st = zk.exists(zkPath, this);
            if (st != null) {
                byte[] b = zk.getData(zkPath, this, st);
                zkValue = new String(b) ;
                log.info(zkPath + "=" + zkValue);
            }
        }

        catch (Exception e1) {
            log.error(e1);
            e1.printStackTrace();
        }

        return zkValue;
    }

    public CassandraConfig getCassandraConfig() {
        return cassandraConfig;
    }

    public String getHostname() {
        return hostname;
    }

    public void setHostname(String hostname) {
        this.hostname = hostname;
    }

    public String getPort() {
        return port;
    }

    public void setPort(String port) {
        this.port = port;
    }

    public int getTimeout() {
        return timeout;
    }

    public void setTimeout(int timeout) {
        this.timeout = timeout;
    }

    public void setCassandraHostPath(String cassandraHostPath) {
        this.cassandraHostPath = cassandraHostPath;
    }

    public void setCassandraPortPath(String cassandraPortPath) {
        this.cassandraPortPath = cassandraPortPath;
    }

    public void setCassandraClusterPath(String cassandraClusterPath) {
        this.cassandraClusterPath = cassandraClusterPath;
    }

    public void setCassandraKeyspacePath(String cassandraKeyspacePath) {
        this.cassandraKeyspacePath = cassandraKeyspacePath;
    }

    public String getMqUrlPath() {
        return mqUrlPath;
    }

    public void setMqUrlPath(String mqUrlPath) {
        this.mqUrlPath = mqUrlPath;
    }

    public String getMqTopicPath() {
        return mqTopicPath;
    }

    public void setMqTopicPath(String mqTopicPath) {
        this.mqTopicPath = mqTopicPath;
    }

    public MQConfig getMqConfig() {
        return mqConfig;
    }


    @Override
    public void process(WatchedEvent event) {
        //To change body of implemented methods use File | Settings | File Templates.
    }
}
